package com.kafka.test.broker.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ExecutionException;

@Component
@Slf4j
public class KafkaService {

  @Resource
  private KafkaConfig kafkaConfig;

  @Resource
  private AdminClient adminClient;

  /**
   * 是否初始化完毕
   */
  private boolean kafkaInitOk = false;

  /**
   * kafka admin client
   *
   * @return
   */
  @Bean
  public AdminClient adminClient() {
    Map<String, Object> config = new HashMap<>();
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBootstrapServers());
    config.put(AdminClientConfig.CLIENT_ID_CONFIG, kafkaConfig.getClientId());
    return AdminClient.create(config);
  }

  /**
   * kafka topic empty
   *
   * @return
   */
  private Collection<TopicListing> topicIsEmpty() {
    ListTopicsResult listTopicsResult = adminClient.listTopics();
    try {
      return listTopicsResult.listings().get();
    } catch (InterruptedException | ExecutionException e) {
      e.printStackTrace();
    }
    return null;
  }


  /**
   * close kafka adminClient
   */
  public void closeAdminClient() {
    if (this.adminClient != null) {
      this.adminClient.close();
      this.adminClient = null;
    }
  }

  /**
   *
   */
  @PostConstruct
  public void createTopic() {
    synchronized (this) {
      try {
        Collection<TopicListing> topicListings = topicIsEmpty();
        if (topicListings != null && topicListings.isEmpty()) {
          List<Topic> topics = kafkaConfig.getTopics();
          if (topics != null && topics.size() != 0) {
            Collection<NewTopic> tmpNewTopics = new ArrayList<>();
            topics.forEach(topic ->
              tmpNewTopics.add(new NewTopic(
                topic.getName(),
                topic.getNumPartition(),
                topic.getReplication())));
            topics.clear();
            CreateTopicsResult createTopicsResult = adminClient.createTopics(tmpNewTopics);
            Map<String, KafkaFuture<Void>> kafkaFutureMap = createTopicsResult.values();
            kafkaFutureMap.forEach((topicName, value) -> {
              log.info("topic {} created ", topicName);
            });
            tmpNewTopics.clear();
          }
        } else {
          if (topicListings != null) {
            for (TopicListing topicListing : topicListings) {
              log.info("topic info:{}", topicListing.toString());
            }
          }
        }
      } finally {
        this.closeAdminClient();
        this.kafkaInitOk = true;
        this.notifyAll(); //block
      }
    }
  }

  /***
   *  kafka init done
   * @return
   */
  private boolean read() {
    synchronized (this) {
      if (!this.kafkaInitOk) {
        try {
          this.wait();
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
      return this.kafkaInitOk;
    }
  }


  private Properties producerConfig() {
    Properties config = new Properties();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBootstrapServers());
    config.put(ProducerConfig.ACKS_CONFIG, "all");//leader and followers ack
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, new StringSerializer());
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, new StringSerializer());



    return config;
  }

  private Properties consumerConfig() {

    Properties config = new Properties();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBootstrapServers());
    config.put(ProducerConfig.ACKS_CONFIG, "all");//leader and followers ack
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, new StringSerializer());
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, new StringSerializer());



    return config;
  }
}
